package com.jhhe.homework.SparkMW


import akka.actor.{Actor, ActorSystem, Props}

import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
import scala.collection.mutable

// 定义sparkMaster节点
class SparkMaster extends Actor {
  //定义个全局的map,管理workers
  private val workers: mutable.Map[String, WorkerInfo] = mutable.Map[String,WorkerInfo]()
  override def receive: Receive = {
    case "start" => {
      println("master服务器启动了。。。")
      // 启动定时信息检测worker是否心跳超时
      self ! StartTimeOutWorker
    }
    case RegisterWorkerInfo(id,cpu,ram) =>{
      if(!workers.contains(id)){
        val workerInfo : WorkerInfo = new WorkerInfo(id,cpu,ram)
        workers += ((id,workerInfo))
        // 如果某一个worker注册成功，那么就给它回复一条信息
        sender() ! RegisteredWorkerInfo
      }
    }
    case HeartBeat(id) =>{
      // 首先将发送心跳的那个worker找出来，然后更新它的心跳时间
      val info : WorkerInfo = workers(id)
      info.lastHeartBeat=System.currentTimeMillis()
      println("master更新了 " + id + " 心跳时间...")
    }

    case StartTimeOutWorker => {
      import context.dispatcher  //导入隐式转换
      println("开始了定时检测worker心跳的任务")
      //说明
      //1. 0 millis 不延时，立即执行定时器
      //2. 9000 millis 表示每隔9秒执行一次
      //3. self:表示发给自己
      //4. RemoveTimeOutWorker 发送的内容
      context.system.scheduler.schedule(0 millis, 9000 millis, self, RemoveTimeOutWorker)
    }

    case RemoveTimeOutWorker =>{
      //取出map中的所有worker
      val workerInfos = workers.values
      // 获取当前系统时间
      val nowTime = System.currentTimeMillis()
      // 对那些 now - lastHeartBeat > 6000 的worker从map中删除
      workerInfos.filter(workerInfo => nowTime-workerInfo.lastHeartBeat>6000)
        .foreach(workerInfo=>workers.remove(workerInfo.id))
      println("当前有 " + workers.size + " 个worker存活的")
    }
  }
}

object SparkMasterApp{
  def main(args: Array[String]): Unit = {
    //这里我们分析出有3个host,port,sparkMasterActor
    if (args.length != 3) {
      //127.0.0.1 10005 SparkMaster
      println("请输入参数 host port sparkMasterActor名字")
      sys.exit()
    }
    val host = args(0)
    val port = args(1)
    val name = args(2)

    //先创建ActorSystem
    val config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider="akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname=${host}
         |akka.remote.netty.tcp.port=${port}
            """.stripMargin)
    val actorSystem = ActorSystem("SparkMasterApp",config)
    // 创建master引用
    val sparkMasterRef = actorSystem.actorOf(Props[SparkMaster],s"${name}")
    sparkMasterRef ! "start"
  }
}
